package com.xwaiy.xwaiyrpc.registry;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import com.xwaiy.xwaiyrpc.config.RegistryConfig;
import com.xwaiy.xwaiyrpc.exception.RpcException;
import com.xwaiy.xwaiyrpc.exception.enums.ErrorCode;
import com.xwaiy.xwaiyrpc.model.ServiceMetaInfo;
import io.etcd.jetcd.*;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.vertx.core.impl.ConcurrentHashSet;

import java.nio.charset.StandardCharsets;
import io.etcd.jetcd.watch.WatchEvent;
import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @Author Xwaiy
 * @Date 2025/8/26 11:41
 **/
public class EtcdRegistry implements Registry {

    private Client client;

    private KV kv;

    private static final String ETCD_ROOT_PATH = "/rpc";

    // 本地注册的节点
    private final Set<String> localRegisterNodes = new HashSet<>();

    // 缓存
    private final RegistryServiceCache regSerCache = new RegistryServiceCache();;

    // 监听
    private final Set<String> watching = new ConcurrentHashSet<>();

    @Override
    public void init(RegistryConfig config) {
        client = Client.builder()
                .endpoints(config.getAddress())
                .connectTimeout(Duration.ofMillis(config.getTimeout()))
                .build();
        kv = client.getKVClient();
        // 开启心跳检测
        heartBeat();
    }

    @Override
    public void register(ServiceMetaInfo serviceMetaInfo) throws Exception {
        Lease leaseClient = client.getLeaseClient();

        // 30秒租约
        long leaseId = leaseClient.grant(30).get().getID();

        String registryKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodekey();
        ByteSequence key = ByteSequence.from(registryKey, StandardCharsets.UTF_8);
        ByteSequence value = ByteSequence.from(JSONUtil.toJsonStr(serviceMetaInfo), StandardCharsets.UTF_8);

        PutOption option = PutOption.builder().withLeaseId(leaseId).build();
        kv.put(key, value, option).get();

        localRegisterNodes.add(registryKey);
    }

    @Override
    public void unRegister(ServiceMetaInfo serviceMetaInfo) {
        String key = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodekey();

        kv.delete(ByteSequence.from(key, StandardCharsets.UTF_8));
        localRegisterNodes.remove(key);
    }

    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        // TODO 先读缓存，不存在再去查后写入缓存
        List<ServiceMetaInfo>  list = regSerCache.readCache(serviceKey);

        if (list != null) {
            return list;
        }

        String search = ETCD_ROOT_PATH + serviceKey + "/";

        try {
            // 前缀查询
            GetOption getOption = GetOption.builder().isPrefix(true).build();
            List<KeyValue> keyValues = kv.get(
                            ByteSequence.from(search, StandardCharsets.UTF_8),
                            getOption)
                    .get()
                    .getKvs();
            // 解析服务信息
            List<ServiceMetaInfo> metaInfoList = keyValues.stream().map(
                    keyValue -> {
                        String key = keyValue.getKey().toString(StandardCharsets.UTF_8);
                        watch(key);
                        String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                        return JSONUtil.toBean(value, ServiceMetaInfo.class);
                    }).collect(Collectors.toList());

            regSerCache.writeCache(serviceKey, metaInfoList);
            return metaInfoList;
        } catch (Exception e) {
            throw new RpcException(ErrorCode.REGISTRY_SERVICE_DISCOVERY_FAILED, "获取服务列表失败", e);
        }
    }

    @Override
    public void heartBeat(){
        // TODO 定时检测心跳（暂定10秒）
        CronUtil.schedule("0/10 * * * * ? ",new Task() {

            @Override
            public void execute() {
                for (String key : localRegisterNodes) {
                    try {
                        List<KeyValue> list = kv.get(ByteSequence.from(key,StandardCharsets.UTF_8))
                                .get().getKvs();
                        // 过期
                        if(CollUtil.isEmpty(list)) {
                            continue;
                        }
                        // 重新注册
                        KeyValue info = list.get(0);
                        String value = info.getValue().toString(StandardCharsets.UTF_8);
                        ServiceMetaInfo metaInfo = JSONUtil.toBean(value, ServiceMetaInfo.class);

                        register(metaInfo);
                    }catch (Exception e){
                        throw new RpcException(ErrorCode.REGISTRY_HEARTBEAT_FAILED, key+"续签失败", e);
                    }
                }
            }
        });

        CronUtil.setMatchSecond(true);
        CronUtil.start();
    }

    @Override
    public void watch(String nodeKey) {
        Watch wclient = client.getWatchClient();

        boolean isWatch = watching.add(nodeKey);
        if (isWatch) {
            wclient.watch(ByteSequence.from(nodeKey,StandardCharsets.UTF_8),resp -> {
                for (WatchEvent event : resp.getEvents()) {
                    switch (event.getEventType()) {
                        case DELETE:
                            // 从nodeKey中提取serviceKey，格式为：/rpc/serviceKey/host:port
                            String[] parts = nodeKey.split("/");
                            if (parts.length >= 3) {
                                String serviceKey = parts[2];
                                regSerCache.cleanCache(serviceKey);
                            }
                            break;
                        case PUT:
                        default:
                            break;
                    }
                }
            });
        }
    }

    @Override
    public void destroy() {
        System.out.println("当前节点下线");

        for (String key : localRegisterNodes) {
            try{
                kv.delete(ByteSequence.from(key,StandardCharsets.UTF_8)).get();
            }catch (Exception e) {
                throw new RpcException(ErrorCode.REGISTRY_SERVICE_UNREGISTER_FAILED, key + "节点下线失败", e);
            }
        }

        // 释放资源
        if (kv != null) {
            kv.close();
        }
        if (client != null) {
            client.close();
        }
    }

}
